package com.hjc.demo.flink.kafka.consumer;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema;

import java.util.Properties;

public class FlinkKafkaConsumerDemo {

    private static String BOOTSTRAP_SERVERS  = "1.15.51.162:9092";
    public static final String GROUP_ID = "test-consumer-group-id";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer(CommonProperties.TOPIC, new SimpleStringSchema(), getProperties());
        environment.addSource(flinkKafkaConsumer).filter(temp ->{
            JSONObject jsonObject = JSONObject.parseObject(temp);
            String srcIp = jsonObject.getString("src_ip");
            return srcIp.equals("1.12.3.50");
        }).print();

        environment.execute("consuemr-kafka-data");


    }

    private static Properties getProperties() {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
        properties.setProperty("group.id", GROUP_ID);
        System.getProperties();
        return properties;

    }
}
